package com.huatai.datacenter.job;

import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.service.ClusterService;
import com.huatai.datacenter.service.KafkaMonitorService;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

/**
 * 程序启动时定时调用kafka监控，结果存入缓存，减少前端调用时间
 *
 * @author lizhong
 * @date 2023年7月28日 16点07分
 */
@Component
@Slf4j
public class KafkaMonitorJob {

	// 新版本，有broker字段的 2023年8月7日
	@Autowired
	private KafkaMonitorService kafkaMonitorService;

	@Autowired
	private ClusterService clusterService;

	/**
	 * 为所有集群clusterID生成缓存，TODO 合理设置定时任务时间！！！！！！！！！！
	 */
	@XxlJob("callListAllConsumeStatKafka")
	public void callListAllConsumeStatKafka() {
		log.info("为所有集群clusterID生成缓存 XXL-JOB:callListAllConsumeStatKafka");
		System.out.println("XXL-JOB: callListAllConsumeStatKafka");

		// 获取所有clusterID 且 state=1
		List<Long> clusterList = clusterService.list().stream().filter(i -> i.getStatus() == 1).
			map(ClusterEntity::getId).distinct().collect(Collectors.toList());

		for (Long clusterID : clusterList) {
			long start = System.currentTimeMillis();
			// 调用，计算多个cluster的，结果存入缓存
			List records = kafkaMonitorService.listAllConsumeStatKafka(String.valueOf(clusterID)).getRecords();
			kafkaMonitorService.setLagStatVOListCache(String.valueOf(clusterID), records);
			long end = System.currentTimeMillis();
			System.out.println("计算集群" + clusterID + "的主题情况列表耗时" + (end - start) + "ms");
		}
	}
}
